package com.ayu.shortlink.project.schedule;

import cn.hutool.core.util.StrUtil;
import com.ayu.shortlink.project.common.constant.ShortLinkConstant;
import com.ayu.shortlink.project.common.convention.exception.ServiceException;
import com.ayu.shortlink.project.dao.entiry.ShortLinkGotoDO;
import com.ayu.shortlink.project.dao.mapper.ShortLinkGotoMapper;
import com.ayu.shortlink.project.dao.mapper.ShortLinkMapper;
import com.ayu.shortlink.project.mq.basics.event.HotKeySaveCaffeineEvent;
import com.ayu.shortlink.project.mq.producer.HotKeySaveToCaffeineProducer;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.redisson.api.RHyperLogLog;
import org.redisson.api.RedissonClient;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.List;

import static com.ayu.shortlink.project.common.constant.ShortLinkConstant.UNION_SET_KEY;
import static com.ayu.shortlink.project.common.constant.ShortLinkConstant.UV_HYPERLOGLOG_KEY;

/**
 * 定时把聚合数据持久化到DB
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class ScheduleDataSaveToDBTask {

    private final StringRedisTemplate stringRedisTemplate;
    private final RedissonClient redissonClient;
    private final ShortLinkMapper shortLinkMapper;
    private final TransactionTemplate transactionTemplate;
    private final HotKeySaveToCaffeineProducer hotKeySaveToCaffeineProducer;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    // 为5000时为hotkey
    private final Integer HOT_PV_NUMBER = 5000;

    // 每3分钟执行一次执行一次 从Redis获取聚合数据然后持久化到DB
    @Scheduled(cron="0 */3 * * * *")
    public void execute(){
        List<String> unionSet;
        try {
            unionSet = stringRedisTemplate.opsForSet().pop(UNION_SET_KEY, 50);
        }catch (Exception exception){
            // 日志打印
            log.warn("Redis服务器异常");
            throw new ServiceException("从Redis获取聚合数据失败");
        }

        // 需要添加 避免空指针
        if (unionSet != null){
            // 格式为   短URL_原URL
            unionSet.forEach(item -> {
                if (StrUtil.isNotBlank(item)){
                    String[] split = item.split("_");
                    String fullShortUrl = split[0];
                    String originUrl = split[1];
                    // 组装获取PV的key
                    String PvKey = String.format(ShortLinkConstant.PV_UNION_KEY, fullShortUrl, originUrl);
                    String pvStr = stringRedisTemplate.opsForValue().get(PvKey);
                    Integer pv;
                    if (StrUtil.isNotBlank(pvStr)){
                        pv = Integer.parseInt(pvStr);
                    } else {
                        pv = 0;
                    }
                    // 删除PV的key
                    stringRedisTemplate.delete(PvKey);

                    // 获取HyperLogLog
                    RHyperLogLog<Object> hyperLogLog = redissonClient.getHyperLogLog(String.format(UV_HYPERLOGLOG_KEY, fullShortUrl));
                    // 获取uv
                    long count = hyperLogLog.count();

                    transactionTemplate.executeWithoutResult((status) -> {
                        try{
                            LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                                    .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                            ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);

                            shortLinkMapper.incrementStats2(shortLinkGotoDO.getGid(), fullShortUrl, pv, (int) count, (int) count);
                        }catch (RuntimeException exception){
                            // 进行回滚 并报警
                            status.setRollbackOnly();
                            log.warn("MySQL聚合插入异常");
                            // 将PV补充回Redis
                            stringRedisTemplate.opsForValue().set(PvKey, String.valueOf(pv));

                            throw new ServiceException("MySQL聚合插入异常");
                        }
                    });

                    // 通过PV判断是否为热key 就比如3分钟点击5000次就为热key(根据实际业务进行调整)
                    if (pv >= HOT_PV_NUMBER){
                        //那么需要通知去服务器去缓存hotkey
                        HotKeySaveCaffeineEvent message = HotKeySaveCaffeineEvent.builder()
                                .fullShortUrl(fullShortUrl)
                                .originUrl(originUrl)
                                .build();

                        SendResult sendResult = hotKeySaveToCaffeineProducer.sendMessage(message);

                        // 发送失败的话 需要进行日志打印
                        if (!sendResult.getSendStatus().equals(SendStatus.SEND_OK)){
                            log.warn("RocketMQ发送hotkey保存到Caffeine失败");
                        }
                    }
                }
            });
        }
    }
}
